package com.ctyun.dws

import com.shujia.utils.SparkTool
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}

object DwsCityTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._

    // 读取dwi层的位置数据融合表

    val mergeLocDF: DataFrame = spark
      .table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)

    // 读取dim层的用户画像表
    val userTagDF: Dataset[Row] = spark.table("dim.dim_usertag_msk_d")
      .where($"day_id" === day_id)

    // 读取行政区划配置表
    val adminCodeDF: DataFrame = spark.table("dim.dim_admincode")

    /**
     * 市游客的定义：
     * 1、停留时间超过30分钟
     * 2、出行距离大于10KM
     */
    mergeLocDF
      // 关联行政区划配置表通过区县id获取对应的城市id
      .join(adminCodeDF.hint("broadcast"), "county_id")

      /**
       * 统计每个用户到每个城市的停留时间
       * 取进入城市的最小的起始时间 再取离开城市的最大的结束时间
       * 计算 时间间隔 作为停留时间
       */
      .withColumn("min_start_date", min($"start_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("max_end_date", max($"end_date") over Window.partitionBy($"mdn", $"city_id"))
      .withColumn("stay_time", dateDiff($"min_start_date", $"max_end_date"))
      .where($"stay_time" > 30 * 60)

      /**
       * 统计每个用户的每个居住地到每个城市的最远出行距离 取最大值作为出行距离
       * 可以通过两个网格编号计算距离：calculateLengthByGrid
       */
      .join(userTagDF, "mdn")
      .withColumn("distance", calculateLengthByGrid($"resi_grid_id", $"grid_id"))
      .withColumn("max_distance", max($"distance") over Window.partitionBy($"mdn", $"resi_grid_id", $"city_id"))
      .where($"max_distance" > 10 * 1000)
      // 根据表结构 整理需要的数据
      .select($"mdn"
        , $"resi_county_id" as "source_county_id"
        , $"city_id" as "d_city_id"
        , round($"stay_time" / 3600, 2) as "d_stay_time"
        , round($"max_distance" / 1000, 2) as "d_max_distance")
      .distinct()
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_city_tourist_msk_d/day_id=$day_id")

    // 增加分区
    spark.sql(
      s"""
         |alter table dws.dws_city_tourist_msk_d  add if not exists partition(day_id='$day_id')
         |""".stripMargin)

  }
}
